package com.Hsu.mq.mqserver.datacenter;

/**
 * @author Hsu琛君珩
 * @ClassName:MemoryDataCenter
 * @date 2024-02-17
 * @apiNote
 * @Version: v1.0
 */

import com.Hsu.mq.common.MqException;
import com.Hsu.mq.mqserver.core.Binding;
import com.Hsu.mq.mqserver.core.Exchange;
import com.Hsu.mq.mqserver.core.MSGQueue;
import com.Hsu.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用这个类统一管理内存中的所有数据
 * 该类后序提供的一些方法，可能会在多线程下使用，因此要注意多线程安全问题
 * 这个类里很涉及很多增删改查给上层 brokerServer 调用，可能同一时刻处理很多请求，即多线程，因此不用 HashMap
 */
public class MemoryDataCenter {
    //exchangeName - Exchange
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //queueName - MSGQueue
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //exchangeName - (queueName - Binding)
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //messageId - Message
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //queueName - Message的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //queueName - (messageId - Message)
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    //针对交换机
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功！exchangeName="+exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！exchangeName="+exchangeName);
    }

    //针对队列
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功！queueName="+queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功！queueName="+queueName);
    }

    //针对绑定
    public void insertBinding(Binding binding) throws MqException {
        //先使用 exchangeName 查一下对应的哈希表是否存在，不存在就创建一个
        //ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        //if(bindingMap==null){
        //    bindingMap=new ConcurrentHashMap<>();
        //    bindingsMap.put(binding.getExchangeName(),bindingMap);
        //}
        //下面这个代码就是以上代码的简化版。这里是线程安全的
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), key -> new ConcurrentHashMap<>());
        //再根据 queueName 查一下，如果已经存在就抛出异常，不存在才能插入。下面两个操作是存在线程不安全的问题的，因此要加锁
        synchronized (bindingMap){//加锁
            if(bindingMap.get(binding.getQueueName())!=null){
                throw new MqException("[MemoryDataCenter] 绑定已经存在！exchangeName="+binding.getExchangeName()+", queueName="+binding.getQueueName());
            }
            //针对数据进行进一步插入
            bindingMap.put(binding.getQueueName(),binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功！exchangeName="+binding.getQueueName()+", queueName="+binding.getQueueName());
    }
    /**
     * 获取绑定两个版本
     * 1.根据 exchangeName 和 queueName 确定唯一一个 Binding
     * 2.根据 exchangeName 获取所有的 Binding
     */
    public Binding getBinding(String exchangeName,String queueName){
        //这里只是进行查询操作，不必加锁
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap==null){
            return null;
        }
        return bindingMap.get(queueName);
    }
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.get(binding.getExchangeName());
        if(bindingMap==null){
            //该交换机没有绑定任何队列，报错
            throw new MqException("[MemoryDataCenter] 绑定不存在！exchangeName="+binding.getExchangeName()+", queueName"+binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功！exchangeName="+binding.getQueueName()+", queueName="+binding.getQueueName());
    }

    /**
     * 添加消息
     */
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功！messageId="+message.getMessageId());
    }

    /**
     * 根据 id 查询消息
     */
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    /**
     * 根据 id 删除消息
     */
    public void removeMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功！messageId="+messageId);
    }

    /**
     * 发送消息到指定队列
     */
    public void sendMessage(MSGQueue queue,Message message){
        //把消息放到指定的数据结构 queueMessageMap
        //根据队列名找到该队列对应的消息链表
        //LinkedList<Message> messages=queueMessageMap.get(queue.getName());
        //if(messages==null){
        //    messages=new LinkedList<>();
        //    queueMessageMap.put(queue.getName(),messages);
        //}
        //下面这个代码就是以上代码的简化版。这里是线程安全的
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), key -> new LinkedList<>());
        //再把数据加到 messages 里面，这里是线程不安全的
        synchronized (messages){//加锁
            messages.add(message);
        }
        //在这里也把消息往消息中心插入。假设 message 已经在消息中心了，重复插入也没关系
        //主要是相同的 messageId，对应的 message 内容也是一样的，服务器代码不会对 Message 内容做修改，这个内容是 basicProperties 和 body
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中！messageId="+message.getMessageId());
    }

    /**
     * 从队列中取消息
     */
    public Message pollMessage(String queueName){
        //根据队列名查找一下对应的队列的消息链表。这是线程安全的
        LinkedList<Message> messages=queueMessageMap.get(queueName);
        //如果没找到就是队列中没有任何消息
        if(messages==null){
            return null;
        }
        synchronized (messages){//加锁
            if(messages.size()==0){
                return null;
            }
            //如果链表中有元素，就进行头删
            Message currentMessage=messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出！messageId="+currentMessage.getMessageId());
            return currentMessage;
        }
    }

    /**
     * 获取指定队列中消息的个数
     */
    public int getMessageCount(String queueName){
        LinkedList<Message> messages=queueMessageMap.get(queueName);
        if(messages==null){
            //队列中没有消息
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    /**
     * 添加未确认的消息
     */
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.computeIfAbsent(queueName,key->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列！messageId="+message.getMessageId());
    }

    /**
     * 删除未确认的消息（消息已经确认了）
     */
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
        if(messageHashMap==null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列中删除！messageId="+messageId);
    }

    /**
     * 获取指定的未确认的消息
     */
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
        if(messageHashMap==null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

    /**
     * 从硬盘上存储数据，把硬盘中之前持久化存储的各个维度的数据都恢复到内存中
     */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0.清空之前所有旧的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //1.恢复所有的交换机数据
        List<Exchange> exchanges=diskDataCenter.selectAllExchanges();
        for(Exchange exchange:exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        //2.恢复所有的队列数据
        List<MSGQueue> queues=diskDataCenter.selectAllQueues();
        for(MSGQueue queue:queues){
            queueMap.put(queue.getName(),queue);
        }
        //3.恢复所有的绑定数据
        List<Binding> bindings=diskDataCenter.selectAllBindings();
        for(Binding binding:bindings){
            ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.computeIfAbsent(binding.getExchangeName(),key->new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        //4.恢复所有的消息数据
        //遍历所有的队列，根据每个队列名字获取到所有的消息
        for(MSGQueue queue:queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for(Message message:messages){
                messageMap.put(message.getMessageId(),message);
            }
        }
        //注意：针对“未确认的消息”，我们不需要从硬盘恢复，之前硬盘存储时也没设定这部分
        //一旦在等待 ack 的过程中，服务器重启了，此时这些“未确认的消息”，就恢复成“未被取走的消息”
        //这个消息在硬盘上存储的时候，就是当做“未被取走”
    }
}